package com.example.demokotlin

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import org.junit.Test


/**
 * 测试流式处理
 *
 * @author zhouronghua
 * @time 2021/9/8 8:56 上午
 */
class FlowTest {

    suspend fun simple(): List<Int> {
        delay(1000) // 假装我们在这里做了一些异步的事情
        return listOf(1, 2, 3)
    }

    @Test
    fun main() = runBlocking<Unit> {
        simple().forEach { value -> println(value) }
    }

    /**============ Stream流 =============**/
    /**
     * 创建流
     *
     * @author zhouronghua
     * @time 2021/9/8 2:32 下午
     */
    private fun simpleFlow(): Flow<Int> = flow {// 流构建器
        println("Flow started")
        for (i in 1 .. 3) {
            delay(100) // 假装我们在这里做了一些有用的事情
            println("Emitting $i")
            emit(i) // 发送下一个值
        }
    }

    @Test
    fun testBlockFlow() = runBlocking<Unit> {
        // 启动并发的协程以验证主线程并未阻塞
        launch {
            for (k in 1 .. 3) {
                println("I'm not blocked $k")
                delay(100)
            }
        }
        // 收集这个流
        simpleFlow().collect { value -> println(value) }
    }

    /**============ Stream流是冷的 =============**/
    /**
     * Flow 是一种类似于序列的冷流 — 这段 flow 构建器中的代码直到流被收集的时候才运行。
     * 这是返回一个流的 simple 函数没有标记 suspend 修饰符的主要原因。
     * 通过它自己，simple() 调用会尽快返回且不会进行任何等待。该流在每次收集的时候启动，
     * 这就是为什么当我们再次调用 collect 时我们会看到“Flow started”。
     */
    @Test
    fun testBlockFlowCold() = runBlocking<Unit> {
        // 启动并发的协程以验证主线程并未阻塞
        println("Calling simple function...")
        val flow = simpleFlow()
        println("Calling collect function...")
        flow.collect { value ->
            println("first collection $value")
        }
        println("Calling collect function again...")
        flow.collect { value ->
            println("again collection $value")
        }
    }

    /**============ Stream流超时取消 =============**/
    /**
     * 流采用与协程同样的协作取消。像往常一样，
     * 流的收集可以在当流在一个可取消的挂起函数（例如 delay）中挂起的时候取消。
     * 以下示例展示了当 withTimeoutOrNull 块中代码在运行的时候流是如何在超时的情况下取消并停止执行其代码的：
     */
    @Test
    fun testBlockFlowCancel() = runBlocking<Unit> {
        // 启动并发的协程以验证主线程并未阻塞
        withTimeoutOrNull(250) {
            simpleFlow().collect { value -> println("collect $value") }
        }
        println("Done")
    }

    /**============ 流构建器flow =============**/
    /**
     * flow { ... } 构建器是最基础的一个。
     * flowOf 构建器定义了一个发射固定值集的流。
     * 使用 .asFlow() 扩展函数，可以将各种集合与序列转换为流。
     */
    @Test
    fun testBlockFlowConstructor() = runBlocking<Unit> {
        // asFlow将一个整数区间转化为Flow流
        (1 .. 10).asFlow().collect { value -> println("collect section flow $value") }
        // flowOf创建的Flow(元素直接通过逗号进行枚举)
        flowOf(1, 2, 3).collect { value -> println("collect flowOf flow $value") }
    }


}